package com.cloud.rocketmqtxprovider.config;

import com.cloud.rocketmqtxprovider.config.listener.RocketMQProviderTransactionListener;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;
import java.util.concurrent.*;

@Configuration
@SuppressWarnings("all")
public class RocketProviderConfig {

    @Autowired
    private RocketMQProviderTransactionListener transactionListener;

    private TransactionMQProducer transactionMQProducer;

    private static final String PROVIDER_GROUP = "ROCKET_TRANSACTION_GROUP";

    private static final String NAME_SERVER_ADDRESS="192.168.142.142:9876";

    //执行任务的线程池
    private final ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 60,
            TimeUnit.SECONDS, new ArrayBlockingQueue<>(50));


    @PostConstruct
    private void init(){
        //初始化事务消息 并添加消息组
        transactionMQProducer = new TransactionMQProducer(PROVIDER_GROUP);
        //设置消息重试次数，这里是指 发送消息的重试次数，如果网络原因导致的重试，会发生消息重复的情况，
        //所以在消费消息的时候，需要判断消息是否已经被消费过了
        transactionMQProducer.setRetryTimesWhenSendFailed(10);
        //设置线程池
        transactionMQProducer.setExecutorService(executor);
        //设置消息的处理监听器，这里编写主要的消息和业务的处理逻辑
        transactionMQProducer.setTransactionListener(transactionListener);
        //设置nameServer的地址
        transactionMQProducer.setNamesrvAddr(NAME_SERVER_ADDRESS);
        //设置发送消息的超时时间
        //transactionMQProducer.setSendMsgTimeout(Integer.MAX_VALUE);
        //开启事务消息
        this.start();
    }

    /**
     * 开启消息
     */
    private void start(){
        try {
            transactionMQProducer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    public TransactionMQProducer getTransactionMQProducer(){
        return transactionMQProducer;
    }


}
